#include "Optimizer.h"


using namespace std;
/*
   Optimizer::Optimizer() : FuncOp(NULL), tList(NULL), andList(NULL),
   groupingAtts(NULL), attsToSelect(NULL),
   distinctAtts(0), distinctFunc(0),
   tables(-1), pipeID(0), opNode(NULL),
   tableNames(NULL), print(0), printPlanFile()
   {}
 */

Optimizer::Optimizer(Statistics & stat,
		struct FuncOperator *functionStruct,
		struct TableList *inptables,
		struct AndList * andL,
		struct NameList * grpAtts,
		struct NameList *  inpAttsToSelect,
		int inpDistinctAtts, int inpDistinctFunc,
		int printScreenFlag, string planFile)
:Stats(stat),
	FuncOp( functionStruct),
	tList(inptables),
	andList(andL), 
	groupingAtts(grpAtts),
	attsToSelect(inpAttsToSelect),
	distinctAtts(inpDistinctAtts),
	distinctFunc(inpDistinctFunc),
	tables(-1),
	pipeID(0),
	root(NULL), 
	tableNames(NULL), 
	print(printScreenFlag),
	printPlanFile(planFile)

{
	if(FuncOp != 0)
		TrimColumn(*FuncOp);
	cout << "Get All Aliases\n";
	tables = GetAllAlias();
	cout << "Base case\n";
	BaseCase();
	#ifdef DEBUG
	cout << "Printing the  Join Vector\n";
	for(int i=0;i<joins.size();i++){
		JoinData tmp = joins.at(i);
		cout << "Name: " << tmp.joinName << " Number: " <<  tmp.numberOfTuples << "\n";
	}
	#endif
	pipeID = 0;
	Schema *tmpSc;
	if(joins.size() > 0){
		cout << "Size > 0\n";
		JoinData tmp = joins[0];
		tmpSc = MakeTree(NULL, NULL, tmp.joinName,-1,-1,0,0,NULL,NULL);	
	} else{
		cout << "Size == 0\n";
		int x;
		string alias = aliasSorted[0].tableName;
		root = CreateSelectNode(alias,&x);
		tmpSc = new Schema("catalog", (char*) tablesMap[alias].c_str());
	}
	cout << "Going to finalize\n";
	FinalizeTree(tmpSc);
}

int Optimizer::GetAllAlias()  //this returns all aliases in sorted order
{
	//filled by yyparse();
	if (tList == NULL)
		return -1;
	else
	{
		struct TableList *temptList = tList;
		//refer TableList structure in yyparse();
		while (temptList != NULL)
		{
			Stats.CopyRel(temptList->tableName, temptList->aliasAs); // copy alias realtion to stats object
			tablesMap[temptList->aliasAs] = temptList->tableName; //for getting bin file using table name
			temptList = temptList->next;
		}
		Stats.Write("tmp.txt");
		for (map <string, string>::iterator map_itr = tablesMap.begin(); map_itr != tablesMap.end(); map_itr++)
		{
			string aliasName = map_itr->first;
			string tableName = map_itr->second;
			SelectData tmp;
			tmp.tableName = aliasName;
			tmp.sel = NULL;
			aliasSorted.push_back(tmp);
		}
		return aliasSorted.size();
	}
}

void Optimizer::BaseCase()
{
	AndList* parseTree = andList;
	AndList* opNode = NULL;
	while(parseTree != NULL)
	{
		OrList* theOrList = parseTree->left;
		while(theOrList != NULL) //to ensure if parse tree contains an OrList and a comparisonOp
		{
			ComparisonOp* compOp = theOrList->left;
			if(compOp == NULL)
				break;
			//first push the left value
			int lCode = compOp ->left->code;
			string lVal = compOp ->left->value;

			//and now the right value
			int rCode = compOp ->right->code;
			string rVal = compOp ->right->value;

			//accept only joins
			if(lCode != NAME || rCode != NAME)
			{
				break;
			}
			size_t leftTable = lVal.find(".");
			size_t rightTable = rVal.find(".");

			string lAlias = lVal.substr(0, leftTable );
			string rAlias = rVal.substr(0, rightTable );
			string joinName = lAlias + "." + rAlias;
			JoinData tmp;
			tmp.joinName = joinName;
			AndList *tmpList = new AndList();
			tmpList->left = theOrList;
			tmpList->rightAnd = NULL;
			tmp.sel = tmpList;
			char** tableNames = new char*[2];
			tableNames[0] = (char*)lAlias.c_str();
			tableNames[1] = (char*)rAlias.c_str();
			tmp.numberOfTuples = Stats.Estimate(tmpList,tableNames,2);
			joins.push_back(tmp);
			theOrList = theOrList->rightOr;
		}
		parseTree = parseTree->rightAnd;
	}

	sort(joins.begin(),joins.end(),JoinCmp());
}

AndList* Optimizer::AndListJoin(vector<string>& aliases)
{
	AndList* outAndList = NULL;
	AndList* parseTree = andList; 
	AndList* opNode = NULL; 
	while(parseTree != NULL)
	{
		cout<<"entered andList join"<<endl;
		bool skipJoin = false;
		OrList* theOrList = parseTree->left;
		while(theOrList != NULL) //to ensure if parse tree contains an OrList and a comparisonOp
		{
			ComparisonOp* compOp = theOrList->left;
			if(compOp == NULL)
				break;
			//first push the left value
			int lCode = compOp ->left->code;
			string lVal = compOp ->left->value;


			//and now the right value
			int rCode = compOp ->right->code;
			string rVal = compOp ->right->value;


			//accept only joins
			if(lCode != NAME || rCode != NAME)
			{
				skipJoin = true;
				break;
			}

			size_t leftTable = lVal.find(".");
			size_t rightTable = rVal.find(".");

			string lAlias = lVal.substr(0, leftTable );
			string rAlias = rVal.substr(0, rightTable );


			bool lMatch = false, rMatch = false;
			for(int i = 0; i < aliases.size(); i++)
			{
				if(aliases.at(i).compare(lAlias) == 0)
				{
					lMatch = true;
				}
				if(aliases.at(i).compare(rAlias) == 0)
				{
					rMatch = true;
				}
			}
			if(!(lMatch && rMatch) )
			{
				skipJoin = true;
				break;
			}


			//move to next orList inside first AND;
			theOrList = theOrList->rightOr;
		}

		opNode = parseTree;
		parseTree = parseTree->rightAnd;
		//if lastAndList was not skipped, append it to newAndList which will be returned
		if(!skipJoin)
		{
			if(opNode == andList)
				andList= parseTree;

			while(outAndList != NULL)
				outAndList = outAndList ->rightAnd;
			outAndList = opNode;
			outAndList ->rightAnd = NULL;
		}
	}
	return outAndList ;
}


AndList* Optimizer::AndListSelection(string alias)
{
	AndList* newAndList = new AndList();
	AndList* tmpAndList = newAndList;
	AndList* lastNode = NULL;
	tmpAndList->left = NULL;
	AndList* parseTree = andList; //copy to ensure boolean is intact
	//cout << "-----------------AndListSelection : " << alias << "--------------------\n";
	while(parseTree != NULL)
	{
		OrList* theOrList = parseTree->left;
		OrList* tmpOr = new OrList();
		OrList* loopOr = tmpOr;
		bool addList = false;
		while(theOrList != NULL) //to ensure if parse tree contains an OrList and a comparisonOp
		{
			ComparisonOp* compOp = theOrList->left;
			if(compOp == NULL) //no list
				break;

			int lCode = compOp ->left->code;
			string ltVal = compOp ->left->value;
			//cout << "Lval: " << lCode << " " << ltVal << "\n";

			int rCode = compOp ->right->code;
			string rVal = compOp ->right->value;
			//cout << "Rval: " << rCode << " " << rVal << "\n";

			//check for non-join
			if(lCode == NAME && rCode == NAME) //skip joins
			{
				break;
			}

			size_t leftTable = ltVal.find(".");
			size_t rightTable = rVal.find(".");
			//cout << "Left Table: " << leftTable << " Right Table: " << rightTable << "\n";
			if(ltVal.compare(0,leftTable , alias) != 0 && rVal.compare(0,rightTable , alias) != 0 )
			{
				theOrList = theOrList->rightOr;
				continue;
			}
			addList = true;
			if(loopOr->left == NULL){
				loopOr->left = compOp;
				loopOr->rightOr = NULL;
			} else {
				loopOr->rightOr = new OrList();
				loopOr = loopOr->rightOr;
				loopOr->left = compOp;
				loopOr->rightOr = NULL;
			}
			theOrList = theOrList->rightOr;
		}
		if(addList == true){
			newAndList->left = tmpOr; 
			newAndList->rightAnd = new AndList();
			lastNode = newAndList;
			newAndList = newAndList->rightAnd;
		} else {
			delete tmpOr;
		}
		parseTree = parseTree->rightAnd;
	}
	if(tmpAndList->left == NULL){
		delete tmpAndList;
		tmpAndList = NULL;
	} else {
		delete lastNode->rightAnd;
		lastNode->rightAnd = NULL;
	}
	return tmpAndList;
}

void Optimizer::makeAndListVect(AndList*  inpParseTree)
{

}

void Optimizer::TrimColumn(FuncOperator& funcOp)
{
	if(funcOp.leftOperand == 0) {
		//cout << "Unexpected unary opeartor\n";
	}

	if(funcOp.leftOperand){
		if(funcOp.leftOperand->code == NAME){
			//cout << "Name: " <<  funcOp.leftOperand->value << "\n";
			string val = funcOp.leftOperand->value;
			int pos =  val.find(".");
			if (pos != string::npos) {
				strcpy(funcOp.leftOperand->value,(char*)val.substr(pos+1).c_str());
			}
			//cout << "Name: " <<  funcOp.leftOperand->value << "\n";
		}
	}
	if(funcOp.leftOperator){
		//cout << "Code: " << funcOp.leftOperator->code << "\n";
	}
	if(funcOp.leftOperator != 0)	
		TrimColumn(*funcOp.leftOperator);
	if(funcOp.right != 0)	
		TrimColumn(*funcOp.right);
}

void Optimizer::TrimColumn(AndList* parseTree)
{
	while(parseTree != NULL)
	{
		OrList* theOrList = parseTree->left;
		while(theOrList != NULL) 
		{
			ComparisonOp* compOp = theOrList->left;
			if(compOp == NULL)
				break;
			int lCode = compOp->left->code;
			string lVal = compOp->left->value;
			if(lCode == NAME)
				strcpy(compOp->left->value, (char*)lVal.substr(lVal.find(".") + 1).c_str()); //returns right of . 										//(only attname)
			//cout<<"left val" << compOp->left->value<<endl;	

			int rCode = compOp ->right->code;
			string rVal = compOp ->right->value;
			if(rCode == NAME)
				strcpy(compOp ->right->value, (char*)rVal.substr(rVal.find(".") + 1).c_str());
			//cout<<"right val" << compOp ->right->value<<endl;

			theOrList = theOrList->rightOr;
		}
		//move to next AndList
		parseTree = parseTree->rightAnd;
	}
}


void Optimizer::GetBestPair(vector<string> & Aliases, AndList* parseTree, 
		pair<string, string> & optimal)
{
	// key being the combination, value being the estimate generated
	map<string, double> allPairEstimate;

	for(int i = 0; i < Aliases.size(); i++)
	{
		pair<string, string> combo;
		combo.first = Aliases.at(i);

		string comboString = "";
		for(int j = 0; j < Aliases.size(); j++)
		{
			if(j == i)
				continue;
			comboString += Aliases.at(j);
			comboString += "."; //add a  '.' marker in between each starting table A.B.C or B.A.C  or 											//C.A.B
		}

		comboString = comboString .substr(0,comboString .size()-1);
		combo.second = comboString ;


		Statistics *tmpStats = new Statistics(*(JoinEstimate[combo.second].stats)); //get it from 								//prestored order estimate

		string orderedCombos = combo.first + "." + combo.second;
		allPairEstimate[orderedCombos] = tmpStats->Estimate(parseTree, tableNames, Aliases.size()*2);
	}

	//now do the min estimated value

	double tuples = -1.0;
	string tempOrder;
	for (map <string, double>::iterator est_itr = allPairEstimate.begin(); est_itr != allPairEstimate.end(); est_itr++)
	{

		if (tuples == -1.0)
		{
			tuples = est_itr->second;
			tempOrder = est_itr->first;
		}
		else
		{
			if(tuples > est_itr->second)
			{
				tuples = est_itr->second;
				tempOrder = est_itr->first;
			}
		}
	}

	int pos = tempOrder .find(".");

	if (pos == string::npos)  //left of first '.' contains starting table name
		cout << "\n error";
	else
	{
		optimal.first = tempOrder .substr(0, pos);
		optimal.second = tempOrder .substr(pos+1);
	}
}


void Optimizer:: MakeQueryPlan()
{
	cout << "=================================================\n";
	cout<<"came to queryplan\n";
	root->PrintNode();
	/* DS to use: joinEstimate map,andListUse map,

	   While doing join, we have to choose best estimate for 2 combo and use that for 3 combo which will be used for final four combo, all these combo's will be present in tableCombo map and estimates are available in joinEstimate map

	   first get the totoal number of tables present
	   1.check for select in each table and create select_file node for each,this is done in GetAllSelections for each alias
	   2. give this as input to the join node as left and right if they have join, otherwise just pass it the topmost node present(this usually doesnot happen)
	   3.now see how many join nodes will be present and create join nodes one above other, 1st,2nd table->first join + 3rd table->2rd join  + 4th table ->3rd join.
	   4. Above this, create a groupby+ sum if both present or sum alone
	   5.above this do the project function
	   6. above this do if distinct if asked

	 */
}
void Optimizer::ExecuteQuery()
{
	cout << "===========================================\n";
	cout<<"came to execute\n";
	cout << "===========================================\n";

	if (root == NULL)
		return;

	clock_t begin = clock();
	root->ExecutePostOrder(); 
	clock_t end = clock();

	double execTime = end - begin;
	double tim = (execTime*1000)/CLOCKS_PER_SEC;

	cout << "Time take for execution: " << double(tim/1000) << " secs \n\n\n";

}

Schema* Optimizer::MakeTree(OperatorNode* left, OperatorNode* right, string joinName, int leftPipe, int rightPipe, int numAttLeft, int numAttRight , Schema* leftSchema,Schema* rightSchema)
{
	string leftAlias = joinName.substr(0,joinName.find("."));
	string rightAlias = joinName.substr(joinName.find(".")+1,joinName.size());
	//cout << "Make tree. " << joinName << " Left: " << leftAlias << " Right: " << rightAlias << "\n";
	if(left == NULL){
		leftPipe = pipeID;
		left = CreateSelectNode(leftAlias,&numAttLeft);
		//cout << "Created Select Node - Left\n";
		leftSchema = new Schema("catalog", (char*) tablesMap[leftAlias].c_str());
	} else{
	}
	if(right == NULL){
		rightPipe = pipeID;
		right = CreateSelectNode(rightAlias,&numAttRight);
		//cout << "Created Select Node - Right\n";
		rightSchema = new Schema("catalog", (char*) tablesMap[rightAlias].c_str());	
	}
	cout <<"LeftPipe: "<< leftPipe <<" Right Pipe: "<< rightPipe <<" Number Left: "<< numAttLeft <<" Number Right: "<< numAttRight <<"\n";

	CNF* myCnf = new CNF();
	Record* inpLit = new Record();
	AndList* outAndList;
	JoinData jtmp;
	int index=0;

	for(int i=0;i<joins.size();i++){
		jtmp = joins[i];
		if(jtmp.joinName == joinName){
			outAndList = jtmp.sel; 
			TrimColumn(outAndList);
			index = i;
		}
	}
	myCnf->GrowFromParseTree(outAndList, leftSchema, rightSchema, *inpLit);
	root = new JoinNode (leftPipe, rightPipe, pipeID++, myCnf, inpLit, numAttLeft, numAttRight);
	root->left = left;	
	root->right = right;	
	//cout << "Created Join Node\n";

	int newNumAtts = numAttLeft + numAttRight;
	FILE *SchemaFile = fopen ("tmp.txt", "w");
	fprintf(SchemaFile , "BEGIN\n%s", joinName.c_str());
	string table=joinName+".tbl"; 
	fprintf(SchemaFile ,"\n%s", table.c_str());
	Attribute * lAtts = leftSchema->GetAtts();
	for (int i = 0; i < leftSchema->GetNumAtts(); i++)
	{
		fprintf(SchemaFile , "\n%s", lAtts [i].name);
		if (lAtts [i].myType == Int)
			fprintf(SchemaFile , " Int");
		else if (lAtts [i].myType == Double)
			fprintf(SchemaFile , " Double");
		else if (lAtts [i].myType == String)
			fprintf(SchemaFile , " String");
	}
	//cout<<"left over"<<endl;
	Attribute * rAtts = rightSchema->GetAtts();
	for (int i = 0; i < rightSchema->GetNumAtts(); i++)
	{
		fprintf(SchemaFile , "\n%s", rAtts [i].name);
		if (rAtts [i].myType == Int)
			fprintf(SchemaFile , " Int");
		else if (rAtts [i].myType == Double)
			fprintf(SchemaFile , " Double");
		else if (rAtts [i].myType == String)
			fprintf(SchemaFile , " String");
	}
	fprintf (SchemaFile , "\nEND\n");
	fclose(SchemaFile );
	delete(leftSchema);
	delete(rightSchema);
	Schema *tmpSchema = new Schema("tmp.txt",(char*)joinName.c_str());

	joins.erase(joins.begin()+index);
	for(int i=0;i<joins.size();i++){
		string jname = joins[i].joinName;
		string lAlias = jname.substr(0,jname.find("."));
		string rAlias = jname.substr(jname.find(".")+1,jname.size());
		cout << "In table: Left " << lAlias << " Right: " << rAlias << "\n";
		if(lAlias == leftAlias || lAlias == rightAlias){
			tmpSchema = MakeTree(root,NULL,jname,pipeID-1,-1,newNumAtts,0,tmpSchema,NULL);
		} else if (rAlias == leftAlias || rAlias == rightAlias){
			tmpSchema = MakeTree(NULL,root,jname,-1,pipeID-1,0,newNumAtts,NULL,tmpSchema);
		}

	}
	return tmpSchema;
}

void Optimizer::FinalizeTree(Schema* tmpSchema){
	Function * pFunc = NULL;
	cout << "Finalizing....\n";
	if (groupingAtts && FuncOp)
	{
		pFunc = new Function();
		cout << "Growing from tree\n";
		pFunc->GrowFromParseTree(FuncOp,*tmpSchema);

		OrderMaker *grpOrder = new OrderMaker();
		NameList* tempGroupingAtts = groupingAtts;
		int maxVal = -1;
		while(tempGroupingAtts != NULL) {
			maxVal++;
			tempGroupingAtts = tempGroupingAtts->next;
		}
		tempGroupingAtts = groupingAtts;
		int* attsToKeep = new int[maxVal+1];
		attsToKeep[0]=0;
		int index=1;
		while(tempGroupingAtts != NULL)
		{
			string attr = string(tempGroupingAtts->name);
			char* name = const_cast<char*>(attr.substr(attr.find(".") + 1).c_str());
			int attIndex = tmpSchema->Find(name);
			cout<<"final at tIndex "<<attIndex <<"\n";
			
			if(attIndex != -1)
			{
				grpOrder->whichAtts[maxVal- grpOrder->numAtts] = attIndex;
				grpOrder->whichTypes[maxVal - grpOrder->numAtts] = tmpSchema->FindType(name);
				grpOrder->numAtts++;
				attsToKeep[index]=attIndex;
				index++;
			}
			tempGroupingAtts = tempGroupingAtts->next;
		}
		int in = pipeID - 1;
		int out = pipeID++;
		int numAttsToKeep=maxVal+2;

		//add sum
		Attribute* schemaAtts = tmpSchema->GetAtts();
		int schemaNumAtts = tmpSchema->GetNumAtts();

		/*Attribute *totAtts = new Attribute[schemaNumAtts + 1];
		totAtts [0].myType = Double;

		for (int i = 1; i <= schemaNumAtts; i++)
		{
			totAtts [i].name = strdup(schemaAtts[i-1].name);
			totAtts [i].myType = schemaAtts[i-1].myType;
		}*/

		string grpFileName = "Group";
                FILE *grpSchemaFile = fopen (grpFileName .c_str(), "w");
                fprintf(grpSchemaFile , "BEGIN\ngroup\ngroup.tbl");
		fprintf(grpSchemaFile , "\n%s", "Sum");
                fprintf(grpSchemaFile , " Double");

                for(int j=0;j<index-2;j++){
			int i = attsToKeep[j];
                        fprintf(grpSchemaFile , "\n%s", schemaAtts[i].name);
                        if (	schemaAtts[i].myType == Int)
                                fprintf(grpSchemaFile , " Int");
                        else if (schemaAtts[i].myType == Double)
                                fprintf(grpSchemaFile , " Double");
                        else if (schemaAtts[i].myType == String)
                                fprintf(grpSchemaFile , " String");
                }

                fprintf (grpSchemaFile , "\nEND\n");
                fclose(grpSchemaFile );


		//Set the schema
	       	Schema * grpSch = new Schema("Group", "group");
		cout << "AttstoKeep: " << numAttsToKeep << "\n";
		OperatorNode* grpNode = new GroupByNode(in, out, pFunc, grpOrder,attsToKeep,grpSch,numAttsToKeep-1,schemaNumAtts+1);
		grpNode->left = root; // make join the left child of group-by
		root = grpNode;
		//updat the attribute to select with the new sum
		attsToSelect = NULL;
	}
	//only sum
	else if (FuncOp)
	{
		//TrimColumn(*FuncOp);
		pFunc = new Function();
		pFunc->GrowFromParseTree (FuncOp, *tmpSchema);

		int in = pipeID-1;
		int out = pipeID++;

		bool checkPrintForProject = false;
		if (attsToSelect == NULL)
			checkPrintForProject = true;

		if (checkPrintForProject && print == 0)
		{
			// schema for projection
			//write only the sum to the file  if print to screen is 0 and no attr to select
			string projFileName = "SumFinal";
			FILE *finalSchemaFile = fopen (projFileName.c_str(), "w");
			fprintf(finalSchemaFile , "BEGIN\n%s\n", "finalSum");
			fprintf(finalSchemaFile , "\nSum Double");
			fprintf (finalSchemaFile , "\nEND\n");
			fclose(finalSchemaFile );
			checkPrintForProject = false;
		}
		OperatorNode * sumNode = new SumNode(in, out, pFunc, checkPrintForProject);
		sumNode->left = root;
		root = sumNode;
	}
	
  /*      if (distinctAtts == 1)
        {
                //Schema * projSch = new Schema("final", "final");
                int in = root->OutPipe;
                int out = pipeID++;
                // Make node for distinct
                OperatorNode * duplicate = new DuplicateNode(in, out, tmpSchema,print);
                duplicate->left = root;
                root = duplicate;
		attsToSelect=NULL;
        }
*/


	if (attsToSelect)
	{
		cout << "In this loop - attsToSelect ==========================\n";
		int in = root->OutPipe;
		int out = pipeID++;
		int numAttsToSelect = 0, numTotAtts = 1;
		vector <int> vec_AttsToKeepNum;
		vector <string> vec_AttsToKeepName;
		vector <Attribute *> vec_dis_atts;

		NameList * temp = attsToSelect;
		while (temp != NULL)
		{
			numAttsToSelect++;
			vec_AttsToKeepName.push_back(temp->name);
			temp = temp->next;
		}

		numTotAtts = tmpSchema->GetNumAtts();

		// iterate over att-to-keep vector and see if that attribute belongs to this table
		// if it does, save its number (offsetted number)
		for (int j = 0; j < vec_AttsToKeepName.size(); j++)
		{
			// remove alias name from the column name
			string fName = vec_AttsToKeepName.at(j);
			string  colName;
			int pos = fName.find(".");
			if (pos == string::npos)
				colName = fName;
			else
				colName = fName.substr(pos+1);

			// find column name in this schema
			int foundPos = tmpSchema->Find((char*)colName.c_str());
			if (foundPos != -1)
			{
				vec_AttsToKeepNum.push_back(foundPos);
                                cout<<"\nfound pos of atts\n"<<foundPos; 
				// for use by project/distinct schema making
				Attribute * pAtt = new Attribute;
				pAtt->name = new char[colName.length()];
				strcpy(pAtt->name, (char*)colName.c_str());
				pAtt->myType = tmpSchema->FindType((char*)colName.c_str());
				vec_dis_atts.push_back(pAtt);
			}
		}

		// convert vector vec_AttsToKeepNum to int*[]
		int size = vec_AttsToKeepNum.size();
		int *attsToKeep = NULL;
		if (size != 0)
			attsToKeep = new int[size];

		for (int j = 0, i = size-1; i >= 0; j++, i--)
			attsToKeep[j] = vec_AttsToKeepNum.at(i); // will reverse
		// Make schema for projection
		string projFileName = "final";
		FILE *projSchemaFile = fopen (projFileName .c_str(), "w");
		fprintf(projSchemaFile , "BEGIN\nfinal\nfinal.tbl");
		int numAtts = vec_dis_atts.size();
		for (int i = 0; i < numAtts; i++)
		{
			Attribute * pAtts = vec_dis_atts.at(i);
			fprintf(projSchemaFile , "\n%s", pAtts->name);
			if (pAtts->myType == Int)
				fprintf(projSchemaFile , " Int");
			else if (pAtts->myType == Double)
				fprintf(projSchemaFile , " Double");
			else if (pAtts->myType == String)
				fprintf(projSchemaFile , " String");
		}

		fprintf (projSchemaFile , "\nEND\n");
		fclose(projSchemaFile );

		int PrintConsole = print;
		if (distinctAtts == 1)  //wil write it to file
			PrintConsole = 0;
		Schema * projSch = new Schema("final", "final");
		OperatorNode  *projNode = new ProjectNode(in, out, attsToKeep, numAttsToSelect,
				numTotAtts, projSch, PrintConsole);
               cout<<"\nprojectdone\n"<<endl;
		projNode->left = root;
		// Make this node top most
		root = projNode;
	}

	if (distinctAtts == 1)
	{
		Schema * projSch = new Schema("final", "final");
		int in = root->OutPipe;
		int out = pipeID++;
		// Make node for distinct
		OperatorNode * duplicate = new DuplicateNode(in, out, projSch,print);
		duplicate->left = root;
		root = duplicate;
	}

	// Print in file
	if (print == 0)
	{
		Schema * projSch = new Schema("final", "final");
		int in = root->OutPipe;
		// Make node for writeout
		OperatorNode * writeOutNode = new WriteOutNode(in, printPlanFile, projSch);
		writeOutNode->left = root;
		root = writeOutNode;
		cout << "\n output to file given by user : " << printPlanFile.c_str() << "\n\n";
	}

}

OperatorNode * Optimizer::CreateSelectNode(string alias, int *num)
{
	string inFile = tablesMap.find(alias)->second;
	int outPipe = pipeID++;

	Schema schema("catalog", (char*)inFile.c_str()); //search for tablename with catalog
	*num = schema.GetNumAtts(); 
	inFile = inFile + ".bin";
	AndList * outAndList = AndListSelection(alias); //searches for all the select predicate with this alias
	OperatorNode * opNode = NULL;
	CNF* inpCNF;
	Record* inpLit;
	if (outAndList != NULL)
	{
		inpCNF = new CNF();
		inpLit = new Record();
		TrimColumn(outAndList);
		inpCNF->GrowFromParseTree(outAndList, &schema, *inpLit);
	}
	else
	{
		cout << "--------------outAndList is null----------------------\n";
		//return all records 
		string attName;
		Attribute * attsList = schema.GetAtts();       
		attName = attsList[0].name; 
		Operand opStruct;
		opStruct.code = NAME;
		opStruct.value = (char*)attName.c_str();
		//build left=right
		ComparisonOp compStruct;
		compStruct.code = EQUALS;
		compStruct.left = &opStruct;
		compStruct.right = &opStruct;

		OrList orStruct;
		orStruct.left = &compStruct;
		orStruct.rightOr = NULL;

		AndList andStruct;
		andStruct.left = &orStruct;
		andStruct.rightAnd = NULL;

		inpCNF = new CNF();
		inpLit = new Record();
		inpCNF->GrowFromParseTree(&andStruct, &schema, *inpLit);
	}
	OperatorNode *tmp = new SelectFileNode(inFile, outPipe, inpCNF, inpLit);
	return tmp;
}

Optimizer::~Optimizer()
{
}


